package com.yang.consumer.kafka.manual;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;

public class KafkaConsumerRebalanceDemo {
    public static final String bootstrapServers = "192.168.137.31:9092,192.168.137.32:9092,192.168.137.33:9092";
    public static final String topic = "topic-demo";
    public static final String groupId = "consumer-demo";
    public static final String clientId = "consumer-demo-1";
    public static final AtomicBoolean isRunning = new AtomicBoolean(true);

    public static Properties initConfig() {
        Properties properties = new Properties();
        // kafka集群broker地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // key反序列化器
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // value序列化器
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 消费组 group.id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // 客户端id 不设置，默认 consumer-1，consumer-2
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, clientId);
        // 手动提交位移
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        // 找不到消费位移，位移越界.重置位移 默认：latest  可选：earliest latest none（抛异常）
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
        return properties;
    }

    /**
     *  暂定方法 consumer.pause();
     *  恢复方法 consumer.resume();
     *  停止消费 consumer.wakeup();
     * @param args
     */
    public static void main(String[] args) {
        Properties properties = initConfig();
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);

        Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
        // 订阅 多次订阅，最后一次为准 subscribe方式订阅会自动再均衡
        // 或者onPartitionsRevoked保存到数据库 onPartitionsAssigned读取数据库seek
        consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener() {
            // 再均衡开始前，消费者停止读取消息后调用
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                consumer.commitSync(currentOffsets);
                currentOffsets.clear();
            }
            // 重新分配分区后，消费者开始读取消息前调用
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> collection) {

            }
        });

        try {
            while (isRunning.get()) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord record : consumerRecords) {
                    System.out.println(record.toString());
                    currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset() + 1));
                }
                consumer.commitSync();
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }


}
